package com.thering;

import com.google.gson.Gson;
import com.thering.bean.CellData;
import com.thering.bean.ReturnResult;
import com.thering.bean.RowData;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;

/**
 * author: the_ring
 * date: 2022/5/27
 * function: HBase operate Class
 */
public class HBaseData {

    private Connection connection = null;
    private Table table = null;
    private ResultScanner scanner = null;


    // construct method
    public HBaseData() throws IOException {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.rootdir", "hdfs://ring1:8020/hbase");
        conf.set("hbase.zookeeper.quorum", "ring1:2181,ring2:2181,ring3:2181");
        conf.set("hbase.client.retries.number", "3");
        this.connection = ConnectionFactory.createConnection(conf);
    }

    // the method will be recall when die
    @Override
    protected void finalize() throws Throwable {
        if (scanner != null) scanner.close();
        if (table != null) table.close();
        if (connection != null) connection.close();

    }

    // according need to return scan
    private Scan getScan(String startRow, String endRow, String family, String qualifier) {
        Scan scan = null;
        if (startRow != null && endRow == null) {
            Get get = new Get(startRow.getBytes(StandardCharsets.UTF_8));
            scan = new Scan(get);
        }
        if (scan == null) {
            scan = new Scan();
        }
        if (endRow != null) {
            scan.withStopRow(Bytes.toBytes(endRow));
        }
        if (family != null && qualifier != null) {
            scan.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier));
        }
        return scan;
    }

    // get data
    private String getData(String tableName, String startRow, String endRow, String sFamily, String sQualifier) {
        // each row data
        List<RowData> rows = new ArrayList<>();

        try {
            // get table instance
            table = connection.getTable(TableName.valueOf(tableName));

            // get ResultScanner
            Scan scan = getScan(startRow, endRow, sFamily, sQualifier);
            scanner = table.getScanner(scan);

            // get Data
            for (Result result : scanner) {
                byte[] row = result.getRow();

                // create row data
                List<CellData> cellDataList = new ArrayList<>();
                RowData rowData = new RowData(new String(row), cellDataList);

                Cell[] cells = result.rawCells();
                for (Cell cell : cells) {
                    byte[] family = CellUtil.cloneFamily(cell);
                    byte[] qualifier = CellUtil.cloneQualifier(cell);
                    byte[] value = CellUtil.cloneValue(cell);

                    cellDataList.add(new CellData(new String(family), new String(qualifier), ByteBuffer.wrap(value).getInt()));
                }

                rows.add(rowData);
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (scanner != null) scanner.close();
            try {
                if (table != null) table.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        // result
        ReturnResult returnResult = new ReturnResult();

        if (!rows.isEmpty()) {
            returnResult.setCode("200");
            returnResult.setMessage("get data successfully");
            returnResult.setResult(rows);
        } else {
            returnResult.setCode("400");
            returnResult.setMessage("some error occurred when get data");
        }

        return new Gson().toJson(returnResult);
    }

    // get all date
    public String getAllData(String tableName) {
        return getData(tableName, null, null, null, null);
    }

    // get a row by row_key
    public String getDataByRowKey(String tableName, String rowKey) {
        return getData(tableName, rowKey, null, null, null);
    }

    // get a date where row_key in a range
    public String getRangeData(String tableName, String startRowKey, String endRowKey) {
        return getData(tableName, startRowKey, null, null, null);
    }

    // get specify column data
    public String getColumnData(String tableName, String family, String qualifier) {
        return getData(tableName, null, null, family, qualifier);
    }

    // get specifier column and specify roe data
    public String getColumnRowData(String tableName, String rowKey, String family, String qualifier) {
        return getData(tableName, rowKey, rowKey, family, qualifier);
    }

    // get specifier column range row data
    public String getColumnRangeRowData(String tableName, String startRow, String endRow, String family, String qualifier) {
        return getData(tableName, startRow, endRow, family, qualifier);
    }

    // get row_key
    public List<String> getRowKey(String tableName) {
        List<String> rowKeys = new ArrayList<>();

        try {
            // get table instance
            table = connection.getTable(TableName.valueOf(tableName));

            // get ResultScanner
            Scan scan = new Scan();
            scanner = table.getScanner(scan);

            // get row_key
            for (Result result : scanner) {
                byte[] row = result.getRow();
                rowKeys.add(new String(row));
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (scanner != null) scanner.close();
            try {
                if (table != null) table.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        return rowKeys;
    }

    public static void main(String[] args) throws IOException {
        HBaseData hBaseData = new HBaseData();
//        String allData = hBaseData.getAllData("movie_type_count");
//        System.out.println(allData);
        System.out.println("-------row data ------");
        String rowData = hBaseData.getDataByRowKey("movie_type_count", "2022-05-25");
        System.out.println(rowData);
//        String movie_type_count = new HBaseData().getColumnRowData("movie_type_count", "2022-06-01", "cf", "动画");
//        System.out.println(movie_type_count);
    }
}
